﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading.Tasks;

namespace Zhaoxi.Protobuf
{
	public class Producer
	{
		public async Task Show()
		{

			string brokerList = "39.96.82.51:9093,47.95.2.2:9092,39.96.34.52:9092";
			string topicName = "test";

			var config = new ProducerConfig
			{
				BootstrapServers = brokerList,
				//幂等性
				EnableIdempotence = true,
				Acks = Acks.All,
				//LingerMs = 10000,
				BatchNumMessages = 1,
				//MessageSendMaxRetries=3,



				//Partitioner = Partitioner.Murmur2Random

			};
			NewInfo newInfo = new NewInfo()
			{
				P_ID = 1,
				P_Title = "tltle",
				P_Content = "P_Content",
				P_CreateTime = DateTime.Now.ToString("yyyyMMdd")
			};
			using (var producer = new ProducerBuilder<string, NewInfo>(config)
				.SetValueSerializer(new CustomStringSerializer<NewInfo>())
				.SetStatisticsHandler((o, json) =>
				{
					Console.WriteLine("json");
					Console.WriteLine(json);
				})
				.Build())
			{

				Console.WriteLine("\n-----------------------------------------------------------------------");
				Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");
				Console.WriteLine("-----------------------------------------------------------------------");
				try
				{
					var deliveryReport = await producer.ProduceAsync(
						topicName, new Message<string, NewInfo> { Key = "newinfo", Value = newInfo });

					Console.WriteLine($"delivered to: {deliveryReport.TopicPartitionOffset}");
				}
				catch (ProduceException<string, string> e)
				{
					Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}]");
				}
			}
		}
	}
}
